package com.ibm.cps.spark.streaming.adapter;

import backtype.storm.task.IOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.ibm.cps.message.RawDataMessage;

import java.util.Map;

/**
 * Created by telekinesis on 4/28/15.
 */
public class TransformBolt implements IRichBolt {
    private IOutputCollector collector;

    public TransformBolt(IOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
    }

    @Override
    public void execute(Tuple tuple) {
        String value1 = tuple.getString(0);
        RawDataMessage message = (RawDataMessage)tuple.getValue(1);
        String str = new String(message.getMessageBytes()) + "_processed";
        collector.emit(SparkAdapterConstants.TRANSFORMED_TOPIC, null, new Values(value1, str));
    }

    @Override
    public void cleanup() {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
